package com.lxb.demo3

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.bridge.scala._
import org.apache.flink.table.api.Expressions.$
import org.apache.flink.table.api.{Table, TableResult}
import org.apache.flink.types.Row
import org.apache.flink.util

/**
 * @author: albert
 * @date: 2021/6/1 16:36
 * @description:
 */
object TableSQLTest1 {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env)

    val dataStream: DataStream[GDP] = env.readTextFile("D:\\project\\bigdata\\demoall\\flinkDemo\\src\\main\\resources\\gdp.txt")
      .map(data => {
        val strings: Array[String] = data.split(",")
        GDP(strings(0), strings(1).toInt, strings(2).toInt, strings(3).toDouble, strings(4).toInt)
      })

    val table: Table = dataStream.toTable(tableEnv)

    val result: TableResult = tableEnv.sqlQuery(s"select  * from $table").execute()

    result.collect()
//    val result: TableResult = tableEnv.executeSql(s"select sum(ods) as num from $table")
//    val value: util.CloseableIterator[Row] = result.collect()
//    while(value.hasNext){
//      val row = value.next
//      println(row)
//    }

    result.print()
    env.execute()


  }
}

case class GDP(province:String,ods:Int,gdp:Int,people:Double,people_avg_gdp:Int)